package org.example.com.atguigu.day05;

import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Iterator;
import java.util.Properties;

public class WriterToMysql {
    public static void main(String[] args) {
        // 从mysql中读数据, 作为数据来源
        SparkSession spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate();
        String url = "jdbc:mysql://hadoop102:3306/gmall?useUnicode=true&characterEncoding=UTF-8";
        String tableName = "user_info";
        Properties props = new Properties();
        props.setProperty("user", "root");
        props.setProperty("password", "000000");
        Dataset<Row> ds = spark.read().jdbc(url, tableName, props);

        // 假设ds2是我们想要保存到mysql中的数据
        Dataset<Row> ds2 = ds.filter("id<50");

        String url2 = "jdbc:mysql://hadoop102:3306/test?useUnicode=true&characterEncoding=UTF-8";
        /*
        ds2.write()
                .mode(SaveMode.Overwrite)
                .jdbc(url2, "testtest", props);  //它会自动建表,自动识别字段类型
                */
        // TODO 工作中写数据到mysql一般不会采用SaveMode.Overwrite, 因为表中有很多历史数据,会直接删除掉,这是不允许的
        // TODO 工作中写数据到mysql一般不会采用SaveMode.Append, 因为可能会导致主键冲突;除非表中没有主见,可以采用这个方式
        // TODO 工作中写数据到mysql一般采用foreachPartition + Insert Into 表名 values(字段值,...) on duplicate key update 字段名=字段值,字段名=字段值,...
        ds2.javaRDD().foreachPartition(new VoidFunction<Iterator<Row>>() {
            @Override
            public void call(Iterator<Row> rowIterator) throws Exception {
                Connection connection = null;
                PreparedStatement statement = null;
                try {
                    connection = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/test?useUnicode=true&characterEncoding=UTF-8", "root", "000000");
                    statement = connection.prepareStatement("Insert Into 表名 values(字段值,...) on duplicate key update 字段名=字段值,字段名=字段值,...");
                    int i = 0;
                    while (rowIterator.hasNext()){
                        Row row = rowIterator.next();
                        statement.setString(1, row.getAs("列名"));
                        if (i%1000==0){
                            statement.executeBatch();
                            statement.clearBatch();
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (statement!=null) {
                        statement.close();
                    }
                    if (connection!=null){
                        connection.close();
                    }
                }
            }
        });
    }
}
